package com.sjgs.gis.fs.store;


import com.sjgs.common.utils.DirUtils;
import com.sjgs.common.utils.FileNameUtils;
import com.sjgs.gis.common.FileSystemFactory;
import com.sjgs.gis.domain.Chunk;
import com.sjgs.gis.domain.DataFile;
import com.sjgs.gis.domain.enumeration.ChunkStatus;
import com.sjgs.gis.domain.enumeration.DataFileType;
import com.sjgs.gis.fs.ChunkStore;
import com.sjgs.gis.fs.FSStore;
import com.sjgs.gis.service.ChunkService;
import com.sjgs.gis.errors.InvalidChunkException;
import com.sjgs.gis.errors.LocalFileIOException;
import com.sjgs.gis.utils.QETag;
import com.sjgs.gis.utils.RegionUtils;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.io.FileSystemResource;
import org.springframework.core.io.Resource;
import org.springframework.core.io.support.ResourceRegion;
import org.springframework.http.HttpHeaders;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;

import java.io.*;
import java.nio.channels.FileChannel;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @author jxw
 * @apiNote 临时文件后端存储服务
 */
@Service
public class LocalFileStore implements FSStore, ChunkStore {

    private static final Log logger = LogFactory.getLog(LocalFileStore.class);
    private static String FILE_ROOT_DIR = "/files";

    @Value("${filesystem.rootDir:/data}")
    private String rootDir;

    @Value("${filesystem.checkRetry:2}")
    private int checkRetry;

    @Value("${filesystem.writeRetry:5}")
    private int writeRetry;

    @Autowired
    ChunkService chkservice;

    public LocalFileStore() {
        this.registryFS();
    }

    @Override
    public void registryFS() {
        logger.info("registry local file system to store providers");
        FileSystemFactory.registry(DataFileType.LOCAL.name(), LocalFileStore.class);
    }

    @Override
    public boolean mkdir(String dir) {
        if (StringUtils.isBlank(dir) || dir.length() < 2) {
            return false;
        }
        DirUtils.mkDir(dir);
        return false;
    }

    @Override
    public boolean rmdir(String dir) {
        if (StringUtils.isBlank(dir) || dir.length() < 2) {
            return false;
        }
        if (Files.isWritable(Paths.get(dir)) && Files.isDirectory(Paths.get(dir))) {
            logger.info("path not exist,create path: " + dir);
            try {
                FileUtils.deleteDirectory(new File(dir));
                return true;
            } catch (IOException e) {
                throw new LocalFileIOException(e.getMessage());
            }
        }
        return false;
    }

    @Override
    public boolean createFile(String path, String content) {
        String dir = DirUtils.getFileDir(path);
        DirUtils.mkDir(dir);
        File file = new File(path);
        if (file.canWrite()) {
            try {
                FileWriter write = new FileWriter(path);
                write.write(content);
                return true;
            } catch (IOException e) {
                throw new LocalFileIOException(e.getMessage());
            }
        }
        return false;
    }

    @Override
    public boolean remove(String path) {
        if (!Files.isWritable(Paths.get(path))) {
            try {
                return Files.deleteIfExists(Paths.get(path));
            } catch (IOException e) {
                throw new LocalFileIOException(e.getMessage());
            }
        }
        return false;
    }

    @Override
    public boolean rename(String oldName, String newName) {
        if (Files.exists(Paths.get(newName))) {
            try {
                FileUtils.forceDelete(new File(newName));
            } catch (IOException e) {
                throw new LocalFileIOException("The File " + oldName + "can't rename");
            }
        }
        new File(oldName).renameTo(new File(newName));
        return false;
    }

    @Override
    public List<String> getFiles(String dir) {
        File file = new File(dir);
        if (file.isDirectory()) {
            return Arrays.asList(file.listFiles())
                    .stream()
                    .map(File::getPath)
                    .collect(Collectors.toList());
        } else {
            return new ArrayList<>();
        }
    }

    @Override
    public boolean pushFile(String sourcePath, String targetPath) {
        return false;
    }

    @Override
    public boolean copyFile(String sourcePath, String targetPath) {
        if (Files.isWritable(Paths.get(sourcePath)) &&
                Files.isWritable(Paths.get(targetPath)) &&
                Files.notExists(Paths.get(targetPath))) {
            File source = new File(sourcePath);
            File target = new File(targetPath);
            if (source.isFile() && target.isFile()) {
                try {
                    org.apache.commons.io.IOUtils.copy(new FileInputStream(source), new FileOutputStream(target));
                } catch (IOException e) {
                    throw new LocalFileIOException(e.getMessage());
                }
                return target.exists() && target.length() > 0 && target.canRead();
            }
        }
        return false;
    }


    @Override
    public Mono<String> mergeChunks(DataFile datafile, Long chunkcount) {
        String outDir = rootDir + FILE_ROOT_DIR + DirUtils.getDir(datafile.getDir());
        String chunkDir = DirUtils.getFileDir(rootDir, datafile.getDir(), datafile.getFkey());
        String outFilePath = outDir + FileNameUtils.getFileName(datafile.getFkey());
        if (Files.notExists(Paths.get(chunkDir))) {
            throw new InvalidChunkException();
        }
        DirUtils.mkDir(outDir);
        String fHash = FileNameUtils.getFileHash(datafile.getFkey());
        if (Files.isReadable(Paths.get(outFilePath))) {
            try {
                if (new QETag().calcETag(outFilePath).equals(fHash)) {
                    return Mono.just(outFilePath);
                } else {
                    if (Files.isWritable(Paths.get(outFilePath))) {
                        Files.deleteIfExists(Paths.get(outFilePath));
                    } else {
                        return Mono.just("");
                    }
                }
            } catch (IOException e) {
                throw new LocalFileIOException(e.getMessage());
            }
        }

        return chkservice.findAllByFkeyAndStatus(datafile.getFkey(), ChunkStatus.FINISHED.ordinal())
                .sort(Comparator.comparing(Chunk::getIndex))
                .collectList()
                .flatMap(chunks -> {
                    if (chunkcount != chunks.size()) {
                        return Mono.just("");
                    }
                    FileOutputStream out = null;
                    FileChannel outChannel = null;
                    try {
                        out = new FileOutputStream(outFilePath, true);
                        outChannel = out.getChannel();
                        long offset = 0L;
                        for (Chunk chunk : chunks) {
                            String chunkPath = chunkDir + "/" + chunk.getIndex();
                            if (Files.notExists(Paths.get(chunkPath)))
                                break;
                            FileInputStream in = null;
                            FileChannel inChannel = null;
                            try {
                                // 记录新文件最后一个数据的位置
                                File chunkFile = new File(chunkPath);
                                in = new FileInputStream(chunkFile);
                                inChannel = in.getChannel();
                                // 从inChannel中读取file.length()长度的数据，写入outChannel的start处
                                outChannel.transferFrom(inChannel, offset, chunkFile.length());
                                offset += chunkFile.length();
                            } catch (Exception e) {
                                throw new LocalFileIOException(e.getMessage());
                            } finally {
                                in.close();
                                inChannel.close();
                            }
                        }
                        if (offset == datafile.getFsize() &&
                                new QETag().calcETag(outFilePath).equals(fHash)) {
                            return Mono.just(outFilePath);
                        }
                    } catch (FileNotFoundException e) {
                        throw new LocalFileIOException(e.getMessage());
                    } catch (IOException e) {
                        throw new LocalFileIOException(e.getMessage());
                    } finally {
                        try {
                            out.close();
                            outChannel.close();
                        } catch (IOException e) {
                            logger.error(e.getMessage());
                        }
                    }
                    return Mono.just("");
                });
    }

    public Mono<Resource> getResourceByFkey(DataFile df) {
        String outDir = rootDir + FILE_ROOT_DIR + DirUtils.getDir(df.getDir());
        String filePath = outDir + FileNameUtils.getFileName(df.getFkey());
        File file = new File(filePath);
        Long start = System.currentTimeMillis();
        if (file.exists() && file.canRead()) {
            return Mono.<Resource>create(monoSink -> {
                try {
                    FileSystemResource resource = new FileSystemResource(filePath);
                    monoSink.success(resource);
                    logger.info("localfs getResourceByFkey cost time :" + String.valueOf((System.currentTimeMillis() - start)) + "ms");
                } catch (Exception e) {
                    monoSink.error(e);
                    throw new LocalFileIOException(filePath);
                }
            });
        } else {
            //定义GrpcFileResource
            String url = df.getUrl();
            String mirro_url = df.getMirroUrl();
            //TODO get grpc host and port
            //TODO get channal for grpc
            //TODO get bytes from grpc client
            return Mono.empty();
        }
    }

    @Override
    public Mono<ResourceRegion> getRegion(Mono<Resource> resourceMono, HttpHeaders headers) {
        return resourceMono.map(resource -> {
            return RegionUtils.getResourceRegion(resource, headers);
        });
    }

    public String mkChidDir(String path, String childpath) {
        StringBuilder sb = new StringBuilder();
        sb.append(path).append("/").append(childpath);
        //判断uploadFolder/identifier 路径是否存在，不存在则创建
        if (!Files.isWritable(Paths.get(sb.toString()))) {
            logger.info("path not exist,create path: " + sb.toString());
            try {
                Files.createDirectories(Paths.get(sb.toString()));
            } catch (IOException e) {
                throw new LocalFileIOException(e.getMessage());
            }
        }
        return sb.toString();
    }
}
